package cn.datawin.cmppclient.sender.provider;

import cn.datawin.cmppclient.sender.entity.Provider;
import cn.datawin.cmppclient.sender.queue.QueueCache;
import cn.datawin.cmppclient.sender.util.SenderUtil;
import cn.datawin.sms.cmpp2.ProviderSubmitMsg;
import cn.datawin.sms.cmpp2.SmsDetailLog;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.drondea.sms.channel.ChannelSession;
import com.drondea.sms.common.util.CommonUtil;
import com.drondea.sms.message.ILongSMSMessage;
import com.drondea.sms.message.IMessage;
import com.drondea.sms.message.MessageProvider;
import com.drondea.sms.message.cmpp.CmppSubmitRequestMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * 提供发送包的接口
 **/
@Slf4j
public class TcpMessageProvider implements MessageProvider {

    private Provider provider;
    private boolean isHoldStatus;

    public TcpMessageProvider(Provider provider) {

        this.provider = provider;
        QueueCache.MESSAGES_QUEUE.putIfAbsent(provider.getNo(), new ConcurrentLinkedQueue<>());
        log.debug("初始化队列消费者---{}", provider.getNo());
    }


    private SmsDetailLog getSmsDataFromRq() {
		SmsDetailLog poll = QueueCache.MESSAGES_QUEUE.get(provider.getNo()).poll();
        if (poll != null) {
            log.debug("poll---{}", JSON.toJSONString(poll));
        }
        return poll;
    }


    public Provider getProvider() {
        return provider;
    }

    public void setProvider(Provider provider) {
        this.provider = provider;
        if (SenderUtil.isChannelHold(provider)) {
            this.isHoldStatus = true;
        } else {
            this.isHoldStatus = false;
        }
    }

    @Override
    public List<IMessage> getTcpMessages(ChannelSession channelSession) {
        return pullMessageFromTopic(channelSession, provider.getNo());
    }

    @Override
    public void responseMessageMatchFailed(String requestKey, IMessage response) {
        //未匹配成功submit处理

    }

    /**
     * 返回是否存在消息,此处是立即返回，回调是异步执行
     *
     * @param channelNo 通道编号
     * @return
     */
    public List<IMessage> pullMessageFromTopic(ChannelSession channelSession, String channelNo) {

        List<IMessage> requestMsg = new ArrayList<>(2);
        //队列拉取
        // TODO: 2022/11/29

		SmsDetailLog smsDetailLog = getSmsDataFromRq();
        if (smsDetailLog == null) {
            return null;
        }

        //2021.01.28 通道挂起状态，直接返回失败，消费掉mq

        //根据submit对象获取要提交的短信
        log.debug("pullMessageFromTopic-----{}", JSON.toJSONString(smsDetailLog));
        List<IMessage> longMsgToEntity = getSubmitMessage(channelSession, smsDetailLog);
        if (longMsgToEntity == null) {
            return null;
        }
        requestMsg.addAll(longMsgToEntity);
        return requestMsg;

    }

    /**
     * 将submit转成协议需要的对象，并拆分长短信
     *
     * @param channelSession
     * @param smsDetailLog
     * @return
     */
    public List<IMessage> getSubmitMessage(ChannelSession channelSession, SmsDetailLog smsDetailLog) {
        //根据submit对象获取要提交的短信
        IMessage submitRequest = SenderUtil.getTcpSendMsg(getProvider(), smsDetailLog, channelSession);
        if (submitRequest == null) {
            return null;
        }

        try {
            log.debug("longMsgToEntity-----{}", JSON.toJSONString(smsDetailLog));
            List<IMessage> longMsgToEntity = CommonUtil.getLongMsgSlices((ILongSMSMessage) submitRequest,
                    channelSession.getConfiguration(), channelSession.getSequenceNumber());
            if (CollectionUtils.isNotEmpty(longMsgToEntity)) {
                int msgLen = longMsgToEntity.size();
                int[] seqIds = new int[msgLen];
                short[] lens = new short[msgLen];
                int index = 0;
                for (IMessage iMessage : longMsgToEntity) {
                    CmppSubmitRequestMessage cmppSubmitRequestMessage = (CmppSubmitRequestMessage) iMessage;
                    log.debug("+++++++++++++++++++{},{}", cmppSubmitRequestMessage.getSequenceId(), cmppSubmitRequestMessage.getMsgLength());
                    seqIds[index] = cmppSubmitRequestMessage.getSequenceId();
                    lens[index] = cmppSubmitRequestMessage.getMsgLength();
                    index++;
                }
                ProviderSubmitMsg providerSubmitMsg = new ProviderSubmitMsg();
                providerSubmitMsg.setSmsDetailLogId(smsDetailLog.getId());
                providerSubmitMsg.setSeqIds(StrUtil.join(",", seqIds));
                providerSubmitMsg.setLens(StrUtil.join(",", lens));
                providerSubmitMsg.setYmd(smsDetailLog.getYmd());
                log.debug("providerSubmitMsg={}", providerSubmitMsg);
//                SpringContextUtil.getBean(ProducerService.class).sendDbMessage(providerSubmitMsg);
            }
            return longMsgToEntity;
        } catch (Exception e) {
            log.error("pullMessageFromTopic", e);
        }
        return null;
    }

    public boolean isHoldStatus() {
        return this.isHoldStatus;
    }

}
